//===----------------------------------------------------------------------===//
//
//                         BusTub
//
// buffer_pool_manager.cpp
//
// Identification: src/buffer/buffer_pool_manager.cpp
//
// Copyright (c) 2015-2021, Carnegie Mellon University Database Group
//
//===----------------------------------------------------------------------===//

#include "buffer/buffer_pool_manager.h"

#include "common/exception.h"
#include "common/macros.h"
#include "storage/page/page_guard.h"

namespace bustub {

BufferPoolManager::BufferPoolManager(size_t pool_size, DiskManager *disk_manager, size_t replacer_k,
                                     LogManager *log_manager)
    : pool_size_(pool_size), disk_manager_(disk_manager), log_manager_(log_manager) {
  // TODO(students): remove this line after you have implemented the buffer pool manager
  /* throw NotImplementedException(
      "BufferPoolManager is not implemented yet. If you have finished implementing BPM, please remove the throw "
      "exception line in `buffer_pool_manager.cpp`."); */

  // we allocate a consecutive memory space for the buffer pool
  pages_ = new Page[pool_size_];
  replacer_ = std::make_unique<LRUKReplacer>(pool_size, replacer_k);

  // Initially, every page is in the free list.
  for (size_t i = 0; i < pool_size_; ++i) {
    free_list_.emplace_back(static_cast<int>(i));
  }
}

BufferPoolManager::~BufferPoolManager() { delete[] pages_; }

auto BufferPoolManager::NewPage(page_id_t *page_id) -> Page * {
  // select new frame from free list first
  latch_.lock();
  frame_id_t t_frame_id = -1;
  if (!free_list_.empty()) {
    t_frame_id = free_list_.back();
    free_list_.pop_back();
  } else {
    // select from replacer
    if (replacer_->Size() == 0) {
      latch_.unlock();
      return nullptr;
    }
    if (!replacer_->Evict(&t_frame_id)) {
      latch_.unlock();
      return nullptr;
    }
  }

  BUSTUB_ASSERT(t_frame_id != -1, "not return after failing to get new frame\n");
  // dirty evicted page should be written back to disk first
  if (pages_[t_frame_id].is_dirty_) {
    disk_manager_->WritePage(pages_[t_frame_id].page_id_, pages_[t_frame_id].data_);
  }
  page_id_t de_page_id = pages_[t_frame_id].page_id_;
  page_table_.erase(de_page_id);

  // init page content and metadata
  *page_id = AllocatePage();
  pages_[t_frame_id].page_id_ = *page_id;
  pages_[t_frame_id].is_dirty_ = false;
  pages_[t_frame_id].ResetMemory();
  pages_[t_frame_id].pin_count_ = 1;
  replacer_->RecordAccess(t_frame_id);
  replacer_->SetEvictable(t_frame_id, false);

  // put new page-frame record to page table
  page_table_.insert(std::make_pair(pages_[t_frame_id].page_id_, t_frame_id));
  latch_.unlock();
  return &pages_[t_frame_id];
}

auto BufferPoolManager::FetchPage(page_id_t page_id, [[maybe_unused]] AccessType access_type) -> Page * {
  // first check if page is in bufferpool
  latch_.lock();
  auto it = page_table_.find(page_id);
  if (it != page_table_.end()) {
    frame_id_t t_frame_id = it->second;
    pages_[t_frame_id].pin_count_++;
    if (pages_[t_frame_id].pin_count_ - 1 == 0) {
      replacer_->SetEvictable(t_frame_id, false);
    }
    replacer_->RecordAccess(t_frame_id);
    latch_.unlock();
    return &pages_[t_frame_id];
  }

  // pages need to move into bufferpool
  // select new frame from free list first
  frame_id_t t_frame_id = -1;
  if (!free_list_.empty()) {
    t_frame_id = free_list_.back();
    free_list_.pop_back();
  } else {
    // select from replacer
    if (replacer_->Size() == 0) {
      latch_.unlock();
      return nullptr;
    }
    if (!replacer_->Evict(&t_frame_id)) {
      latch_.unlock();
      return nullptr;
    }
  }

  BUSTUB_ASSERT(t_frame_id != -1, "not return after failing to get new frame\n");
  // dirty evicted page should be written back to disk first
  if (pages_[t_frame_id].is_dirty_) {
    disk_manager_->WritePage(pages_[t_frame_id].page_id_, pages_[t_frame_id].data_);
  }

  // delete evicted page record in table
  page_id_t de_page_id = pages_[t_frame_id].page_id_;
  page_table_.erase(de_page_id);

  // read page from disk
  disk_manager_->ReadPage(page_id, pages_[t_frame_id].data_);
  pages_[t_frame_id].is_dirty_ = false;
  pages_[t_frame_id].page_id_ = page_id;
  pages_[t_frame_id].pin_count_ = 1;

  // set record in table and replacer
  replacer_->RecordAccess(t_frame_id);
  replacer_->SetEvictable(t_frame_id, false);
  page_table_.insert(std::make_pair(page_id, t_frame_id));
  latch_.unlock();
  return &pages_[t_frame_id];
}

auto BufferPoolManager::UnpinPage(page_id_t page_id, bool is_dirty, [[maybe_unused]] AccessType access_type) -> bool {
  // get frame id in page table
  latch_.lock();
  auto it = page_table_.find(page_id);
  if (it == page_table_.end()) {
    latch_.unlock();
    return false;
  }
  frame_id_t t_frame_id = it->second;

  // page already unpinned
  if (pages_[t_frame_id].pin_count_ <= 0) {
    latch_.unlock();
    return false;
  }

  // else
  pages_[t_frame_id].pin_count_--;
  // pages_[t_frame_id].is_dirty_ ? 1 : pages_[t_frame_id].is_dirty_ = is_dirty;
  if (!pages_[t_frame_id].is_dirty_) {
    pages_[t_frame_id].is_dirty_ = is_dirty;
  }
  if (pages_[t_frame_id].pin_count_ <= 0) {
    replacer_->SetEvictable(t_frame_id, true);
  }
  latch_.unlock();
  return true;
}

auto BufferPoolManager::FlushPage(page_id_t page_id) -> bool {
  // get frame id
  latch_.lock();
  auto it = page_table_.find(page_id);
  if (it == page_table_.end()) {
    latch_.unlock();
    return false;
  }
  frame_id_t t_frame_id = it->second;
  disk_manager_->WritePage(page_id, pages_[t_frame_id].data_);
  pages_[t_frame_id].is_dirty_ = false;
  latch_.unlock();
  return true;
}

void BufferPoolManager::FlushAllPages() {
  latch_.lock();
  for (auto &pair : page_table_) {
    disk_manager_->WritePage(pair.first, pages_[pair.second].data_);
    pages_[pair.second].is_dirty_ = false;
  }
  latch_.unlock();
}

auto BufferPoolManager::DeletePage(page_id_t page_id) -> bool {
  // if page isn't in bfp,return true
  latch_.lock();
  auto it = page_table_.find(page_id);
  if (it == page_table_.end()) {
    latch_.unlock();
    return true;
  }
  frame_id_t t_frame_id = it->second;

  // if page is pinned, return false
  if (pages_[t_frame_id].pin_count_ > 0) {
    latch_.unlock();
    return false;
  }

  // remove page from replacer and page table
  replacer_->Remove(t_frame_id);
  page_table_.erase(page_id);

  // reset page data and metadata
  pages_[t_frame_id].is_dirty_ = false;
  pages_[t_frame_id].ResetMemory();
  pages_[t_frame_id].page_id_ = INVALID_PAGE_ID;

  DeallocatePage(page_id);

  // add frame into free list
  free_list_.emplace_back(static_cast<int>(t_frame_id));
  latch_.unlock();
  return true;
}

auto BufferPoolManager::AllocatePage() -> page_id_t { return next_page_id_++; }

auto BufferPoolManager::FetchPageBasic(page_id_t page_id) -> BasicPageGuard {
  // std::cout << "FetchPageBasic(" << page_id << ")" << std::endl;
  Page *t_page = FetchPage(page_id);
  return {this, t_page};
}

auto BufferPoolManager::FetchPageRead(page_id_t page_id) -> ReadPageGuard {
  // std::cout << "FetchPageRead(" << page_id << ")" << std::endl;
  Page *t_page = FetchPage(page_id);
  t_page->RLatch();
  return {this, t_page};
}

auto BufferPoolManager::FetchPageWrite(page_id_t page_id) -> WritePageGuard {
  // std::cout << "FetchPageWrite(" << page_id << ")" << std::endl;
  Page *t_page = FetchPage(page_id);
  t_page->WLatch();
  return {this, t_page};
}

auto BufferPoolManager::NewPageGuarded(page_id_t *page_id) -> BasicPageGuard {
  // std::cout << "NewPageGuarded()" << std::endl;
  Page *t_page = NewPage(page_id);
  return {this, t_page};
}

}  // namespace bustub
